package com.desheng.bigdata.flink.stream.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}
import redis.clients.jedis.Jedis

/**
  * 将结果内容统计放入redis
  */
object _02Sink2RedisOps {
    def main(args: Array[String]): Unit = {

        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        val lines = env.fromCollection(List(
            "hello you",
            "hello me",
            "hello she"
        ))


        val pairs: DataStream[(String, Int)] = lines.flatMap(_.split("\\s+"))
                    .map((_, 1))
                    .keyBy(0)
                    .sum(1)

//        pairs.addSink(new MyRedisSinkFunction)//自己写的redissink
        val jedisConfig = new FlinkJedisPoolConfig.Builder()
                            .setDatabase(0)
                            .setHost("bigdata01")
                            .setPort(6379)
                            .build()
        pairs.addSink(new RedisSink(jedisConfig, new MyRedisMapper()))
        env.execute()
    }
}

class MyRedisMapper extends RedisMapper[(String, Int)] {
    /*
        该方法就是对我们要执行的redis操作的说明
            我们要处理的类型(String, list, hash, set, zset)，
     */
    override def getCommandDescription: RedisCommandDescription = {
        new RedisCommandDescription(RedisCommand.HSET, "flink_kafka")
    }

    //我们要从输入类型中提取value，基于getCommandDescription提供的操作进行处理
    override def getValueFromData(data: (String, Int)): String = {
        data._2 + ""
    }

    //我们要从输入类型中提取key,如果是普通的类型，那就是key，但是如果为hash，就为field字段
    override def getKeyFromData(data: (String, Int)): String = {
        data._1
    }
}

class MyRedisSinkFunction extends RichSinkFunction[(String, Int)] {
    var jedis: Jedis = null

    override def invoke(pair: (String, Int), context: SinkFunction.Context[_]): Unit = {
        jedis.hset("flink_kafka", pair._1, pair._2 + "")
    }

    override def open(parameters: Configuration): Unit = {
        jedis = new Jedis("bigdata01", 6379)
    }

    override def close(): Unit = {
        if(jedis != null)
            jedis.close()
    }
}
